/*
package com.kqkj.dyrule.marketing.functions;



import com.kqkj.dyrule.marketing.beans.*;
import com.kqkj.dyrule.marketing.controller.TriggerModelRuleMatchController;
import com.kqkj.dyrule.marketing.utils.RuleSimulatorFromJson;
import com.kqkj.dyrule.marketing.utils.StateDescContainer;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.List;

*/
/**
 * 规则匹配计算核心函数
 *  规则：
 *        触发事件： K事件，事件属性（P2=V1）
 *        画像属性条件：tag87=v2 tag26=v1   hbase
 *        行为次数条件：2022-04-17-当前  事件C[p6=v8,p12=v12] 做过>=2次 clickhouse
 *//*

public class RuleMatchKeyedProcessFunction extends KeyedProcessFunction<String, DynamicKeyedBean, RuleMatchResult> {

    TriggerModelRuleMatchController triggerModelRuleMatchControlller;
    List<MarketingRule> rules;
    ListState<EventBean> listState;
    ListState<Tuple2<MarketingRule, Long>> ruleTimerState;
    @Override
    public void open(Configuration parameters) throws Exception {
        listState = getRuntimeContext().getListState(StateDescContainer.getEventBeansDesc());
        triggerModelRuleMatchControlller = new TriggerModelRuleMatchController(listState);
        rules = RuleSimulatorFromJson.getRule();

        //记录规则定时注册信息的state
        ruleTimerState = getRuntimeContext().getListState(StateDescContainer.ruleTimerStateDesc);




    }

    @Override
    public void processElement(DynamicKeyedBean dynamicKeyedBean,Context context, Collector<RuleMatchResult> collector) throws Exception {
        //把数据流事件放入state
        EventBean eventBean = dynamicKeyedBean.getEventBean();
        listState.add(eventBean);
        for (MarketingRule rule : rules) {
            //查看是否满足触发事件，没触发进行下一次迭代
            //if (!EventUtil.compare(eventBean,rule.getTriggerEventCondition())) continue;//进行下一次迭代
            boolean b = false;
            if (rule.getKeyByFields().equals(dynamicKeyedBean.getKeyNames())){
                //规则匹配计算，看规则的所有条件是否满足
                b = triggerModelRuleMatchControlller.ruleIsMatch(rule, eventBean);
            }
            if (b) {
                //这个规则是否是一个带定时条件的规则
                if (rule.isOnTimer()){
                    //从规则中取出所有定时条件
                    List<TimerCondition> timerConditionList = rule.getTimerConditionList();
                    for (TimerCondition timerCondition : timerConditionList) {
                        //注册定时器
                        context.timerService().registerEventTimeTimer(eventBean.getTimeStamp()+timerCondition.getTimeLate());
                        //在定时信息state进行记录
                        ruleTimerState.add(Tuple2.of(rule,eventBean.getTimeStamp()+timerCondition.getTimeLate()));
                    }
                } else {
                    //不带定时条件
                    RuleMatchResult ruleMatchResult = new RuleMatchResult(eventBean.getKeyByValue(), rule.getRuleId(), eventBean.getTimeStamp(), System.currentTimeMillis());
                    collector.collect(ruleMatchResult);
                }

            }

        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<RuleMatchResult> out) throws Exception {

        Iterable<Tuple2<MarketingRule, Long>> ruleTimerIterable = ruleTimerState.get();
        Iterator<Tuple2<MarketingRule, Long>> iterator = ruleTimerIterable.iterator();
        while(iterator.hasNext()){
            Tuple2<MarketingRule, Long> tp = iterator.next();
            //判断这个规则+定时点，是否对应本次触发点
            if (tp.f1 == timestamp) {
                //如果对应，检查该规则的定时条件(定时条件中包含的就是行为组合条件列表)
                for (TimerCondition timerCondition : tp.f0.getTimerConditionList()) {
                    //调用service去检查再条件指定的时间范围内，事件组合发生次数是否满足
                    boolean b = triggerModelRuleMatchControlller.isMatchTimeCondition(ctx.getCurrentKey(), timerCondition, timestamp - timerCondition.getTimeLate(), timestamp);
                    //清除已经检查完的规则定时点state信息
                    iterator.remove();
                    if (b){
                        RuleMatchResult ruleMatchResult = new RuleMatchResult(ctx.getCurrentKey(), tp.f0.getRuleId(), timestamp - timerCondition.getTimeLate(), ctx.timerService().currentProcessingTime());
                        out.collect(ruleMatchResult);
                    }
                }
            }
            //增加删除过期定时信息的逻辑(一般不会出现)
            if (tp.f1 < timestamp) {
                iterator.remove();
            }
        }
    }
}
*/
